package com.mio.flinkdemo;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import static org.apache.flink.table.api.Expressions.$;

public class FlinktableDemo {

	public static void main(String[] args) throws Exception {
		// Flink执行环境env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// 用env，做出Table环境tEnv
		StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

		// 获取流式数据源
		DataStreamSource<Tuple2<String, Integer>> data = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
			@Override
			public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
				while (true) {
					ctx.collect(new Tuple2<>("name", 10));
					Thread.sleep(1000);
				}
			}

			@Override
			public void cancel() {
			}
		});

		// Table方式
		// 将流式数据源做成Table
		Table table = tEnv.fromDataStream(data, $("name"), $("age"));
		// 对Table中的数据做查询
		Table name = table.select($("name"));
		// 将处理结果输出到控制台
		DataStream<Tuple2<Boolean, Row>> result = tEnv.toRetractStream(name, Row.class);
		// SQL方式:
		/*
		 * tEnv.createTemporaryView("userss",data, $("name"), $("age")); String s =
		 * "select name from userss"; Table table = tEnv.sqlQuery(s);
		 * DataStream<Tuple2<Boolean, Row>> result = tEnv.toRetractStream(table,
		 * Row.class);
		 */
		result.print();
		env.execute();

	}
}
